package com.hkbigdata.window;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.stream.Stream;

/**
 * @author liuanbo
 * @creat 2024-04-25-9:43
 * @see 2194550857@qq.com
 */
public class Flink02_ProcessTime_SlidingWindows {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = env.socketTextStream("hadoop102", 9999)
                .flatMap((String data, Collector<Tuple2<String, Integer>> out) -> {
                            String[] split = data.split(",");
   /*                 for (int i = 0; i < split.length; i++) {
                        out.collect(Tuple2.of(split[i],1));
                    }
                }*/
                            Stream<String> stream = Arrays.stream(split);
                            stream.forEach(word -> out.collect(Tuple2.of(word, 1)));
                        }
                ).returns(Types.TUPLE(Types.STRING, Types.INT));

        flatMap.keyBy(data -> data.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();
    }
}
